package com.yunxi.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.*;
import com.yunxi.pms.cache.service.ICacheManageService;
import com.yunxi.pmsToMQ.pojo.VehiclePdiBillCheckMQVO;
import com.yunxi.pmsToMQ.service.StoPdiCheckToMQService;
import com.yunxi.rocketmq.config.RocketMQConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
import java.io.UnsupportedEncodingException;
import java.util.Properties;
import java.util.Random;

/**
 * Created by caoyong on 2019/09/24.
 */
@Component
public class StoPdiCheckConsumer implements ApplicationListener<ContextRefreshedEvent> {


    private static final Logger logger = LoggerFactory.getLogger(StoPdiCheckConsumer.class);

    @Autowired
    private RocketMQConfig rocketMQConfig;
    @Value("${aliyun.rocketMQ.sto.pdi.topic}")
    private String topic;
    @Value("${aliyun.rocketMQ.sto.pdi.tag}")
    private String tag;
    @Value("${aliyun.rocketMQ.sto.pdi.consumerGroup}")
    private String group;
    @Value("${cache.switch}")
    private Boolean cacheSwitch;

    @Autowired
    @Qualifier(value = "stoPdiCheckToMQServiceHub")
    StoPdiCheckToMQService stoPdiCheckToMQService;

    @Autowired
    ICacheManageService iCacheManageService;


    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        Properties properties = rocketMQConfig.getProperties();
        properties.put(PropertyKeyConst.GROUP_ID, group);
        // 集群订阅方式设置（不设置的情况下，默认为集群订阅方式）
        properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);

        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(topic, tag, new MessageListener() {
            @Override
            public Action consume(Message message, ConsumeContext context) {
                logger.info(tag + " Receive: " + new String(message.getBody()));
                try {
                    String eaiLogs = new String(message.getBody(), "utf-8");
                    JSONObject jsonObj = (JSONObject) JSON.parse(eaiLogs);
                    VehiclePdiBillCheckMQVO vehiclePdiBillCheckMQVO = JSONObject.toJavaObject(jsonObj, VehiclePdiBillCheckMQVO.class);

                    try {
                        stoPdiCheckToMQService.savePdiCheckFromMq(vehiclePdiBillCheckMQVO);
                        if(cacheSwitch){
                            iCacheManageService.putCache(vehiclePdiBillCheckMQVO.getPdiBillId(),true);
                        }
                    } catch (Exception e) {
                        if(cacheSwitch){
                            iCacheManageService.putCache(vehiclePdiBillCheckMQVO.getPdiBillId(),false);
                        }
                        e.printStackTrace();
                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }

                return Action.CommitMessage;
            }
        });

        consumer.start();
        logger.info("消费者启动===========");
    }
}
